package com.atguigu.flink.chapter12;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/21 14:37
 */
public class Flink_SQL_TopN {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 1. 先建一个表与source关联
        tenv.executeSql("create table ub(" +
                            "   user_id bigint, " +
                            "   item_id bigint, " +
                            "   category_id int, " +
                            "   behavior string, " +
                            "   ts bigint, " +
                            "   et as to_timestamp(from_unixtime(ts)), " +
                            "   watermark for et as et - interval '3' second " +
                            ")with(" +
                            "   'connector' = 'filesystem', " +
                            "   'path' = 'input/UserBehavior.csv', " +
                            "   'format' = 'csv' " +
                            ")");
    
        
        // 2. 执行查询
        // 2.0 过滤出来点击纪律
        Table t1 = tenv.sqlQuery("select * from ub where behavior='pv'");
        tenv.createTemporaryView("t1", t1);
        // count(*) count(id) sum(1)
        // 2.1 开窗口(分组窗口) 统计点击量
        Table t2 = tenv.sqlQuery("select " +
                                        " item_id, " +
                                        " hop_start(et, interval '30' minute, interval '1' hour) stt, " +
                                        " hop_end(et, interval '30' minute, interval '1' hour) edt, " +
                                        " count(*) ct " +
                                        "from t1 " +
                                        "group by item_id, hop(et, interval '30' minute, interval '1' hour)");
    
        tenv.createTemporaryView("t2", t2);
        // 2.2 使用over窗口给每个点击量添加排名
        Table t3 = tenv.sqlQuery("select" +
                                        " *, " +
                                        " row_number() over(partition by edt order by ct desc) rn " +
                                        "from t2 ");
    
        tenv.createTemporaryView("t3", t3);
        // 2.3 取出top3
        Table t4 = tenv.sqlQuery("select " +
                                        "edt w_end, " +
                                        "item_id, " +
                                        "ct item_count, " +
                                        "rn rk " +
                                        "from t3 " +
                                        "where rn <= 3");
    
        //3. 建一个表与sink(mysql)关联
        tenv.executeSql("CREATE TABLE `hot_item` (\n" +
                            "  `w_end` timestamp ,\n" +
                            "  `item_id` bigint,\n" +
                            "  `item_count` bigint,\n" +
                            "  `rk` bigint,\n" +
                            "  PRIMARY KEY (`w_end`,`rk`) not enforced\n" +
                            ")with(" +
                            "   'connector' = 'jdbc', " +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/flink_sql', " +
                            "   'table-name' = 'hot_item', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ") ");;
        
        
        // 4. 把2.3的结果写入到sink表
        t4.executeInsert("hot_item");
        
    }
}
